package spark.core.scala

import org.apache.spark.{AccumulableParam, SparkConf, SparkContext}

import scala.collection.mutable._
/**
 * Created by ibf on 2018/2/4.
 */
/**
 * 使用map集合来解决wc的计数问题
 * 将每一次遍历得到的字符串加入map当中，如果map中存在这样的key，就直接往value上叠加1
 *                                  如果map中不存在这样的key，就将(String -> 1)添加到map中
 */
object MyMapAccumulable extends AccumulableParam[Map[String,Int],String] {
  /**
   * 添加Key到map集合中的方法
   * @param r  累加器保持的集合map
   * @param t  每一次要添加的Key：Sting
   * @return  返回处理过后的累加器：map
   */
  override def addAccumulator(r: Map[String, Int], t: String): Map[String, Int] = {
    //往map中添加元素，如果r中有t，就取到t对应的value，否则就返回0，然后跟前面的1进行叠加
    r += t -> (1+r.getOrElse(t,0))
  }

  /**
   * 不同分区的累加器的合并方法
   * @param r1
   * @param r2
   * @return
   */
  override def addInPlace(r1: Map[String, Int], r2: Map[String, Int]): Map[String, Int] = {
      //遍历r1,传入初始值r2，a是r2 ，在r2中判断，是否包含r1的key，如果包含就取出并且跟r1的值做叠加，
      //                                                      如果不包含就返回0然后加上r1的值
      r1.foldLeft(r2)((a,b) => {
        a += b._1 -> (a.getOrElse(b._1,0)+b._2)
      })
  }

  /**
   * 给定初始值
   * @param initialValue
   * @return
   */
  override def zero(initialValue: Map[String, Int]): Map[String, Int] = {
    initialValue
  }
}

object DefineMyAccumulatorWC {
    def main(args: Array[String]) {
      //1、构建spark上下文
      val conf = new SparkConf()
        .setAppName("AccumulatorsWC")
        .setMaster("local[*]")
      val sc = SparkContext.getOrCreate(conf)

      //2、生成RDD
      val rdd = sc.parallelize(Array(
        "hadoop,spark,hbase",
        "spark,spark,hbase",
        "hadoop,spark,hadoop",
        "",
        "hadoop,spark,hbase",
        "hive,spark,hbase",
        "hue,spark,beeline",
        "flume,spark,hbase",
        "oozie,hue,hbase",
        null,
        "hue,spark,kafka"
      ),3)

        /**
         * 需求二：不使用reduceByKey来实现单词统计
         *
         *  实际场景：为了避免reduceByKey或者类似的shuffle聚合操作，会使用累加器
         *            为了统计某些数值的时候，使用累加器，一般来说是简单的计数
         *   使用自定义的累加器来完成wc
         */
      //3、创建自定义的累加器
      val myMapAccumulable = sc.accumulable(Map[String,Int]())(MyMapAccumulable)

      //4、处理数据
      rdd.foreachPartition(iter => {
        iter.foreach(line => {
          val nline = if(line == null)"" else line
          //进行数据分割过滤转换
          nline.split(",")
               .filter(_.trim.nonEmpty)
               .map(word => {
            println(word)
            myMapAccumulable += word
          })
        })
      })


      //5、累加器结果打印(打印map集合)
      println(myMapAccumulable.value)
      myMapAccumulable.value.foreach(println)

      //6、线程等待
      Thread.sleep(100000l)

    }
}
